package com.gt.ln

import com.gt.SCUtil
import org.apache.commons.lang3.StringUtils
import org.apache.spark.SparkContext
import org.apache.spark.broadcast.Broadcast
import org.apache.spark.rdd.RDD



object Spark_hot_product_03 {

  def main(args: Array[String]): Unit = {
    val sc: SparkContext = SCUtil.createLocalSc()

    //1. 获取数据
    val rdd: RDD[String] = sc.textFile("data/user_visit_action.txt")

    //val clickCategoryId = arr(6) //点击类 品类id
    //val bookCategoryId = arr(8) //下单类 品类id a-b-c
    //val payCategoryId = arr(10) //支付类 品类id a-b-c
    //2. 转换数据格式
    val rdd2: RDD[(String, (Int, Int, Int), String)] = rdd.map(line => line.split("_"))
      .map(arr => {
        val sessionId = arr(2)
        val clickCategoryId = arr(6) //点击类 品类id
        val bookCategoryId = arr(8) //下单类 品类id a-b-c
        val payCategoryId = arr(10) //支付类 品类id a-b-c

        if (StringUtils.isNotEmpty(clickCategoryId) && clickCategoryId.toInt != -1 && !"null".equals(clickCategoryId)) {
          (clickCategoryId, (1, 0, 0),sessionId)
        } else if (StringUtils.isNotEmpty(bookCategoryId) && !"null".equals(bookCategoryId)) {
          (bookCategoryId, (0, 1, 0),sessionId)
        } else if (StringUtils.isNotEmpty(payCategoryId) && !"null".equals(payCategoryId)) {
          (payCategoryId, (0, 0, 1),sessionId)
        } else {
          ("", (0, 0, 0),sessionId)
        }
      }).filter(data => StringUtils.isNotEmpty(data._1) && !data._1.equals("null"))

    val rdd2_1: RDD[(String, (Int, Int, Int),String)] = rdd2.flatMap(data => {
        data._1.split(",").map(id => List((id, data._2, data._3)))
      })
      .flatMap(data => data)

    val rdd3: RDD[(String, (Int, Int, Int))] = rdd2_1.map(data => (data._1, data._2))

    val rdd4: RDD[product] = rdd3.reduceByKey((a, b) => {
      (a._1 + b._1, a._2 + b._2, a._3 + b._3)
    }).map(data => {
      product(data._1, data._2._1, data._2._2, data._2._3)
    }).sortBy(p => p)


    val list: List[String] = rdd4.collect().reverse.take(10).map(data => {
      data.id
    }).toList

    val bc: Broadcast[List[String]] = sc.broadcast(list)

    val rdd5: RDD[(String, List[(String, Int)])] = rdd2_1.filter(data => bc.value.contains(data._1) && data._2._1>0)
      .map(data => (data._1 + "_" + data._3, 1))
      .reduceByKey(_ + _)
      .map(data => {
        val arr: Array[String] = data._1.split("_")
        val pid = arr(0)
        val sessionId = arr(1)
        val count = data._2
        (pid, (sessionId, count))
      })
      .groupByKey()
      .map(data => {
        val pid = data._1
        val list1: List[(String, Int)] = data._2.toArray.sortBy(_._2)(Ordering.Int.reverse).take(10).toList
//        val list1: List[(String, Int)] = data._2.toArray.sortWith((a,b) =>{
//          if(a._2>b._2){
//            true
//          }else{false}
//        }).take(10).toList
        (pid, list1)
      })


    rdd5.collect().foreach(println)

    sc.stop()
  }

}
